package cn._51doit.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
/**
 * @create: 2021-10-13 16:18
 * @author: 珠穆朗玛峰顶峰的男人
 * @program: kafka_java_scala
 * 生产者  kafka的ack机制  可以保证生产者写入数据是安全的
 * 有三个参数 0 1 -1
 *     0:一直写不等leader应答   没法保证数据的安全
 *     1:只等leader应答   不等followers同步应答   相对安全   如果leader死机数据有可能丢失
 *     -1/all:不但要leader将数据落盘,还要等待所有的follower分区数据落盘,然后响应leader分区,
 *            最后leader分区响应生产者,生产者才会继续写数据
 **/
public class JavaACK {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // 连接kafka节点
        properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092");
        //指定key反序列化方式
        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //指定value反序列化方式
        properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //设置ack
        properties.setProperty("acks","1");
        //设置retroes
        properties.setProperty("retries", Integer.MAX_VALUE+"");
        //生产者的实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //封装数据
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("word", "hi flink");
        //写入数据的方法
        producer.send(producerRecord);
        //关闭资源
        producer.close();
    }
}
